package com.ruoyi.mqtt.handler;

import com.ruoyi.mqtt.annotation.QpsMonitor;
import com.ruoyi.mqtt.model.SensorData;
import com.ruoyi.mqtt.service.DataProcessingService;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

@Slf4j
@Component
public class MqttMessageHandler implements MessageHandler {
    private static final int BATCH_SIZE = 1000;
    private static final int QUEUE_SIZE = 100000;
    
    private final ObjectMapper objectMapper;
    private final DataProcessingService dataProcessingService;
    private final BlockingQueue<SensorData> messageQueue;
    private volatile boolean running = true;

    public MqttMessageHandler(ObjectMapper objectMapper, DataProcessingService dataProcessingService) {
        this.objectMapper = objectMapper;
        this.dataProcessingService = dataProcessingService;
        this.messageQueue = new ArrayBlockingQueue<>(QUEUE_SIZE);
        startBatchProcessor();
    }

    @Override
    @QpsMonitor(value = "MQTT消息接收", timeWindow = 5)
    public void handleMessage(Message<?> message) throws MessagingException {
        try {
            String payload = message.getPayload() instanceof byte[] ? 
                new String((byte[]) message.getPayload()) : 
                (String) message.getPayload();
                
            SensorData sensorData = objectMapper.readValue(payload, SensorData.class);
            messageQueue.offer(sensorData);
        } catch (Exception e) {
            log.error("处理消息失败", e);
        }
    }

    private void startBatchProcessor() {
        Thread processor = new Thread(() -> {
            List<SensorData> batch = new ArrayList<>(BATCH_SIZE);
            
            while (running) {
                try {
                    messageQueue.drainTo(batch, BATCH_SIZE);
                    
                    if (!batch.isEmpty()) {
                        dataProcessingService.processBatch(batch);
                        batch.clear();
                    } else {
                        Thread.sleep(1);
                    }
                } catch (Exception e) {
                    log.error("批处理失败", e);
                }
            }
        });
        processor.setName("batch-processor");
        processor.start();
    }
} 